package cn.doitedu.eagle.queryservice;

import cn.doitedu.eagle.beans.*;
import cn.doitedu.eagle.names.MetricNames;
import cn.doitedu.eagle.ruleutils.ClickHouseUtil;
import cn.doitedu.eagle.ruleutils.RuleParamUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.metrics.Counter;

import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/3/13
 *
 * 用户行为属性条件查询服务
 * 对时间进行分段双路查询
 **/
@Slf4j
public class UserActionCountQueryServiceImpl implements UserActionCountQueryService {
    SimpleDateFormat sdf;
    // 事件序列state
    ListState<LogBean> eventState;

    // 缓存管理器
    FlinkStateBufferManager flinkStateBufferManager;

    // clickHouse连接
    Connection conn;

    // ck查询metric计数器
    Counter ckQueryCounter;

    // 缓存有效次数metric计数器
    Counter bufferAvailCounter;


    public UserActionCountQueryServiceImpl(RuntimeContext runtimeContext,ListState<LogBean> eventState,
                                           FlinkStateBufferManager flinkStateBufferManager) throws Exception {
        this.sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        this.eventState = eventState;
        this.flinkStateBufferManager = flinkStateBufferManager;
        conn = ClickHouseUtil.getClickHouseConnection();
        ckQueryCounter = runtimeContext.getMetricGroup().counter(MetricNames.CK_QUERY_CNTS);
        bufferAvailCounter = runtimeContext.getMetricGroup().counter(MetricNames.BUFFER_AVAIL_CNTS);
    }

    /**
     * 用户行为属性条件查询入口
     * 针对的条件模板：   count(e1[p1=v1,p2=v2],startTime->endTime) >= n次
     *
     * @param ruleParam
     * @return
     * @throws Exception
     */
    public boolean queryByRangeAndMinEventCounts(RuleParam ruleParam) throws Exception {

        String deviceId = ruleParam.getDeviceId();
        String ruleId = ruleParam.getRuleId();

        long curTime = System.currentTimeMillis();

        // 远近分界点（当前时间-15分钟的前1小时对小时取整）
        long splitPoint = DateUtils.ceiling(DateUtils.addMinutes(new Date(curTime), -75), Calendar.HOUR).getTime();

        // 拿到该规则对应的事件阈值查询sql
        String sqls = ruleParam.getEventCountQuerySqls();
        String[] sqlArr = sqls.split(";");

        // 遍历每一个"行为阈值条件"，进行查询判断
        List<RuleParamEvent> actionJudgeEvents = ruleParam.getEventCountParamList();
        for (int i = 0; i < actionJudgeEvents.size(); i++) {
            // 取到当前预置条件封装参数
            RuleParamEvent eventParam = actionJudgeEvents.get(i);
            Integer judgeCount = eventParam.getJudgeCount();

            // 设置当前阈值条件查询对应的sql
            eventParam.setQuerySql(sqlArr[i]);

            /**
             * 处理时间格式
             * **/
            String timeRangeStart = eventParam.getTimeRangeStart();
            String timeRangeEnd = eventParam.getTimeRangeEnd();

            // 转成long格式
            long queryStartTime = DateUtils.parseDate(timeRangeStart, "yyyy-MM-dd HH:mm:ss").getTime();
            // 如果end是-1，则表示要计算到当前
            long queryEndTime = "-1".equals(timeRangeEnd) ? curTime : DateUtils.parseDate(timeRangeEnd, "yyyy-MM-dd HH:mm:ss").getTime();

            /*
             * 从缓存中查询
             * 缓存中key-value的格式： ( eventParamStr,cnt|endTime )
             * **/
            String bufferKey = RuleParamUtil.ruleEventParam2Str(eventParam);
            BufferResult bufferResult = flinkStateBufferManager.getCountBuffer(bufferKey,queryStartTime,queryEndTime,splitPoint);

            /*
             *  缓存处理后的击穿查询
             *  但是利用了缓存，也减少了击穿后的查询范围
             * **/
            BufferAvailLevel availLevel = bufferResult.getAvailLevel();
            int farCount = 0;
            int nearCount = 0;
            int count = 0;
            switch (availLevel) {
                case UN_AVAIL:
                    log.info("规则：{},阈值缓存无效：{} ,查询的的deviceId：{},缓存返回：{},条件阈值：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {}",ruleId,
                            bufferKey,deviceId,bufferResult.getBufferValue(),judgeCount,queryStartTime,queryEndTime,bufferResult.getBufferStartTime(),bufferResult.getBufferEndTime(),splitPoint);
                    // 完整查询
                    farCount = queryEventCountsFar(ruleId, deviceId, eventParam, queryStartTime, queryEndTime);
                    ckQueryCounter.inc();

                    nearCount = queryEventCountsNear(eventParam, queryStartTime, queryEndTime, eventState);
                    count = farCount+nearCount;

                    // 更新缓存
                    flinkStateBufferManager.updateCountBuffer(bufferKey,count,queryStartTime,queryEndTime);

                    // 只要当前阈值条件不满足，则整个行为次数阈值规则返回false
                    if (count < judgeCount) return false;
                    break;

                case FAR_PARTIAL:
                    log.info("规则：{},阈值缓存FAR_PARTIAL有效：{} ,查询的的deviceId：{},缓存返回：{},条件阈值：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {}",ruleId,
                            bufferKey,deviceId,bufferResult.getBufferValue(),judgeCount,queryStartTime,queryEndTime,bufferResult.getBufferStartTime(),bufferResult.getBufferEndTime(),splitPoint);

                    bufferAvailCounter.inc();
                    // 如果缓存查询的有效结果已经满足阈值，则直接跳过本阈值条件的后续计算
                    if(bufferResult.getBufferValue()>=judgeCount) {
                        log.info("规则：{},阈值缓存FAR_PARTIAL命中：{} ,查询的的deviceId：{},缓存返回：{},条件阈值：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {}",ruleId,
                                bufferKey,deviceId,bufferResult.getBufferValue(),judgeCount,queryStartTime,queryEndTime,bufferResult.getBufferStartTime(),bufferResult.getBufferEndTime(),splitPoint);
                        continue;
                    }

                    // 查询一部分历史时段+near时段
                    farCount = queryEventCountsFar(ruleId, deviceId, eventParam, bufferResult.getShiftStartTime(), bufferResult.getShiftEndTime()) ;
                    ckQueryCounter.inc();

                    nearCount = queryEventCountsNear(eventParam,splitPoint,queryEndTime,eventState);
                    count = bufferResult.getBufferValue() + farCount + nearCount;

                    // 更新缓存
                    flinkStateBufferManager.updateCountBuffer(bufferKey,count,queryStartTime,queryEndTime);

                    if (count < judgeCount) return false;
                    break;

                case FAR_NEAR_PARTIAL:
                    log.info("规则：{},阈值缓存FAR_NEAR_PARTIAL有效：{} ,查询的的deviceId：{},缓存返回：{},条件阈值：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {}",ruleId,
                            bufferKey,deviceId,bufferResult.getBufferValue(),judgeCount,queryStartTime,queryEndTime,bufferResult.getBufferStartTime(),bufferResult.getBufferEndTime(),splitPoint);

                    bufferAvailCounter.inc();

                    // 如果缓存查询的有效结果已经满足阈值，则直接跳过本阈值条件的后续计算
                    if(bufferResult.getBufferValue()>=judgeCount) {
                        log.info("规则：{},阈值缓存FAR_NEAR_PARTIAL命中：{} ,查询的的deviceId：{},缓存返回：{},条件阈值：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {}",ruleId,
                                bufferKey,deviceId,bufferResult.getBufferValue(),judgeCount,queryStartTime,queryEndTime,bufferResult.getBufferStartTime(),bufferResult.getBufferEndTime(),splitPoint);
                        continue;
                    }

                    // 查询一部分near时段
                    nearCount = queryEventCountsNear(eventParam,splitPoint,queryEndTime,eventState);
                    count = bufferResult.getBufferValue() + nearCount;

                    // 更新缓存
                    flinkStateBufferManager.updateCountBuffer(bufferKey,count,queryStartTime,queryEndTime);

                    if (bufferResult.getBufferValue() + nearCount < judgeCount) return false;
                    break;
            }
        }

        // 上面整个for循环走完没有提前返回的话，说明每个条件都吻合，那么返回true
        ruleParam.setHit(true);  // 设置结果到参数对象中
        return true;
    }


    /**
     * 对一个事件条件进行“历史”查询
     *
     * @param eventParam 事件条件的详细参数
     * @param startTime  判断条件的统计起始时间
     * @param endTime    判断条件的统计结束时间
     * @return
     */
    private int queryEventCountsFar(String ruleId, String deviceId, RuleParamEvent eventParam, long startTime, long endTime) throws Exception {


        String ruleSql = eventParam.getQuerySql();
        PreparedStatement pst = conn.prepareStatement(ruleSql);
        pst.setString(1, deviceId);
        // 设置查询起始时间条件
        pst.setLong(2, startTime);
        pst.setLong(3, endTime);

        long qs = System.currentTimeMillis();
        ResultSet resultSet = pst.executeQuery();

        int cnt = 0;
        while (resultSet.next()) {
            cnt = resultSet.getInt(2);
        }
        long qe = System.currentTimeMillis();
        log.info("规则：{},阈值条件ck查询: {},deviceid：{},queryStart：{},queryEnd：{},返回：{},耗时：{} 毫秒",ruleId,RuleParamUtil.ruleEventParam2Str(eventParam),deviceId,startTime,endTime,cnt,qe-qs);
        eventParam.setRealCount(cnt);

        return cnt;
    }


    /**
     * 对一个事件条件进行“近期”查询
     *
     * @param eventParam 事件条件的详细参数
     * @param startTime  判断条件的统计起始时间
     * @param eventState 存放近期行为事件的state
     * @return 吻合条件的事件次数
     * @throws Exception
     */
    private int queryEventCountsNear(RuleParamEvent eventParam, long startTime, long endTime, ListState<LogBean> eventState) throws Exception {

        // 取出状态中的near事件列表
        Iterable<LogBean> logBeans = eventState.get();

        // 定义一个满足条件的次数计数器
        int c = 0;
        // 遍历状态中的每一个事件
        for (LogBean logBean : logBeans) {
            // 如果状态中的事件id == 条件中的事件id，且时间范围吻合，则继续进行属性的判断
            if (logBean.getEventId().equals(eventParam.getEventId()) && logBean.getTimeStamp() >= startTime && logBean.getTimeStamp() <= endTime) {
                // 取出事件条件中的属性条件
                Map<String, String> properties = eventParam.getProperties();
                Set<String> keys = properties.keySet();

                // 遍历属性条件
                boolean flag = true;
                for (String key : keys) {
                    // 如果某个属性条件不满足，则直接跳出循环
                    if (!properties.get(key).equals(logBean.getProperties().get(key))) {
                        flag = false;
                        break;
                    }
                }
                // 如果所有属性条件都吻合，则计数++
                if (flag) c++;
            }
        }

        eventParam.setRealCount(c);
        return c;
    }
}
